package com.cetc.sdp.kmga.cs.stream;

import com.cetc.sdp.kmga.cs.jdbc.DeviceDAO;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.broadcast.Broadcast;
import scala.Tuple2;

import java.util.List;
import java.util.Set;

/**
 * @description:
 * @author： DengQiang
 * @date: 2017/12/13 15:12
 */
public class DeviceSetBroadcastWrapper extends AbstractBroadcastWrapper<Set<String>> {

    private volatile static DeviceSetBroadcastWrapper INSTANCE;

    private DeviceSetBroadcastWrapper(JavaSparkContext jsc, Set<String> devices) {
        super(jsc);
        this.devices = devices;
    }

    private Set<String> devices;

    public static DeviceSetBroadcastWrapper getInstance(JavaSparkContext jsc, Set<String> devices) {
        if (INSTANCE == null) {
            synchronized (DeviceSetBroadcastWrapper.class) {
                if (INSTANCE == null) {
                    INSTANCE = new DeviceSetBroadcastWrapper(jsc, devices);
                }
            }
        }
        return INSTANCE;
    }

    public void addDevice(String dev) {
        if (this.devices.add(dev)) {
            refreshBroadcast();
        }
    }

    public void removeDevice(String dev) {
        if (this.devices.remove(dev)) {
            refreshBroadcast();
        }
    }

    @Override
    protected void reload() {
        List<Tuple2<String, Integer>> devs = DeviceDAO.findAllDevice();
        devices.clear();
        devs.forEach(tuple -> devices.add(tuple._1()));
    }

    @Override
    protected Broadcast<Set<String>> broadcast(JavaSparkContext jsc) {
        return jsc.broadcast(this.devices);
    }
}